package me.tuyou.common;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * 定义数据源
 */
public class WordStreamGenerator implements SourceFunction<String> {

    private boolean isRunning = true;

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        List<String> list = new ArrayList<>();
        list.add("flink");
        list.add("stream");
        list.add("batch");
        list.add("table");
        list.add("sql");

        while (isRunning) {
            int index = new Random().nextInt(list.size());
            sourceContext.collect(list.get(index));
            TimeUnit.SECONDS.sleep(1);
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }
}
